/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package com.wider.foundation.concurrency;

import java.util.Collection;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * This class defines an extension of the ArrayBlockingQueue that supports counting the number of blocked threads.
 * This class of queue is bounded to the capacity specified by the caller during constructing the instance.
 * This class supports implementations of the QueuingAware interface.
 * 
 * @author William Choi <avextk@gmail.com>
 * @see LinkedBlockingQueue
 */
public class CountedBlockingQueue<E> extends LinkedBlockingQueue<E> {
    private int             blocking_count;
    private final String    count_lock;

    
    /**
     * Creates a CountedBlockingQueue with a capacity of Integer.MAX_VALUE.
     */
    public CountedBlockingQueue(){
        super();
        blocking_count  = 0;
        count_lock      = "locker for blocking count";
    }

    /**
     * Creates a CountedBlockingQueue with a capacity of Integer.MAX_VALUE, initially containing the elements of the given collection, added in traversal order of the collection's iterator.
     * @param c     the collection of elements to initially contain
     */
    public CountedBlockingQueue(Collection<? extends E> c){
        super(c);
        blocking_count  = 0;
        count_lock      = "locker for blocking count";
    }

    /**
     * Creates a LinkedBlockingQueue with the given (fixed) capacity.
     * @param capacity  the capacity of this queue. 
     */
    public CountedBlockingQueue(int capacity){
        super(capacity);
        blocking_count  = 0;
        count_lock      = "locker for blocking count";
    }

    /**
     * Inserts the specified element at the tail of this queue if it is possible to do so immediately without exceeding the queue's capacity,
     * returning true upon success and throwing an IllegalStateException if this queue is full.
     * @param e The element to add.
     * @return True if success, false otherwise.
     * @throws NullPointerException
     * @throws IllegalStateException
     */
    @Override
    public boolean add(E e) throws NullPointerException, IllegalStateException {
        boolean ret = false;

        if (super.offer(e)){
            ret = true;
            if (e instanceof QueuingAware){
                ((QueuingAware) e).enterQueue(this);
            }
        }else{
            ret = false;
            throw new IllegalStateException("Queue is full.");
        }
        return ret;
    }

    /**
     * Inserts the specified element at the tail of this queue if it is possible to do so immediately without exceeding the queue's capacity,
     * returning true upon success and false if this queue is full.
     * This method is generally preferable to method add(E), which can fail to insert an element only by throwing an exception.
     * @param e The element to add.
     * @return  True if sucess; false otherwise.
     * @throws NullPointerException
     */
    @Override
    public boolean offer(E e) throws NullPointerException {
        boolean ret = false;
        if (super.offer(e)){
            ret = true;
            if (e instanceof QueuingAware){
                ((QueuingAware) e).enterQueue(this);
            }
        }
        return ret;
    }

    /**
     * Retrieves and removes the head of the queue, waiting if no elements present in this queue. Couting when the reader is blocked.
     * @return  The head element of the queue.
     * @throws InterruptedException
     */
    public E fetch() throws InterruptedException {
        E ret = null;
        
        //Notify all blocking threads (producers) which are waiting for the consumers
        synchronized (this){
            this.notifyAll();
        }

        synchronized (count_lock){
            blocking_count++;
        }
        
        ret = take();
        
        synchronized (count_lock){
            blocking_count--;
        }

        if (ret instanceof QueuingAware){
            ((QueuingAware) ret).exitQueue(this);
        }
        return ret;
    }


    /**
     * Method for user to retrieve the current number of blocking threads on this queue.
     * @return  The number of currently blocked threads
     */
    public int getBlockingCount(){
        return blocking_count;
    }
}
